package highlevel;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.WindowStore;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Arrays;
import java.util.Date;
import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/4/17 10:41
 * @Version: 1.0
 * @description:kafka streaming highlevel 窗口操作 window
 */
public class HighLevelTransationWindow {
    

        public static void main(String[] args) {
          //创建properties对象配置kafka集群以及相关
            Properties properties = new Properties();
            properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");
            properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
            properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"word-dsl");
            properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,3);
            

            //创建StreamBuild对象，可以获取kafkastreaming的topology任务
            StreamsBuilder streamsBuilder = new StreamsBuilder();

            
            //从消费主题进行获取record中的数据
            KStream<String, String> kStream= streamsBuilder.stream("t9");


//            //翻滚（Tumbling window）窗口：有固定的时间间隔，数据不重叠，将流数据通过时间划分为的独立的窗口，微批处理的结果
//            KTable<Windowed<String>, Long> kTable = kStream.flatMapValues(value -> Arrays.asList(value.split(" ")))
//
//                    //根据value的值进行分组
//                    .groupBy((k, v) -> v)
//
//                    //设置每个窗口与每个窗口之间的时间间隔

                        /*
                        * 例：0-5s
                        *     5-10s
                        *     10-15s
                        * */
//                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//
//                    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("AA").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));
//
//                    kStream.flatMapValues(value -> Arrays.asList(value.split(" ")))
//
//                            //根据value的值进行分组
//                            .groupBy((k, v) -> v)
//
//
//                            //设置每个窗口与每个窗口之间的时间间隔
//                            .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//
//                            .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("AA").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));
//
//
//            kTable.toStream().foreach((k,v)->{
//
//                //获取window对象
//                Window window = k.window();
//
//                //获取时间间隔的开始时间
//                long start = window.start();
//
//                //获取时间间隔的结束时间
//                long end = window.end();
//
//                //时间格式化的处理
//                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                String startTime = dateFormat.format(new Date(start));
//                String endTime = dateFormat.format(new Date(end));
//
//                System.out.println(startTime+"\t"+endTime+"\t"+k.key()+"\t"+v);
//
//
//            });




            //跳跃（Hopping window）窗口：有固定的时间间隔，数据重复，按照固定的时间间隔，推进相同的间隔
//            KTable<Windowed<String>, Long> kTable = kStream.flatMapValues(value -> Arrays.asList(value.split(" ")))
//
//                    //根据value的值进行分组
//                    .groupBy((k, v) -> v)
//
//                    /*设置每个窗口与每个窗口之间的时间间隔,同时设置推进的时间间隔
//                    例：0-10s
//                      5-15s
//                      10-20s
//                    */
//                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)).advanceBy(Duration.ofSeconds(5)))
//
//                    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("AA").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));
//
//            kStream.flatMapValues(value -> Arrays.asList(value.split(" ")))
//
//                    //根据value的值进行分组
//                    .groupBy((k, v) -> v)
//
//
//                    //设置每个窗口与每个窗口之间的时间间隔
//                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//
//                    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("AA").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));
//
//
//            kTable.toStream().foreach((k,v)->{
//
//                //获取window对象
//                Window window = k.window();
//
//                //获取时间间隔的开始时间
//                long start = window.start();
//
//                //获取时间间隔的结束时间
//                long end = window.end();
//
//                //时间格式化的处理
//                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                String startTime = dateFormat.format(new Date(start));
//                String endTime = dateFormat.format(new Date(end));
//
//                System.out.println(startTime+"\t"+endTime+"\t"+k.key()+"\t"+v);
//
//
//            });
//


            //会话（Session window）窗口：动态大小，数据不重复，数据驱动窗口（随着数据的添加窗口的时间，自动进行有规律的递增）

            /*
            * servlet中的session的默认生命周期为30分钟，如果在生命周期未到时有使用，自动续斻生命周期30分以此类推
            *
            * session的移除只有两种方式：1.Session超时机制，2：手动移除
            * */

            KTable<Windowed<String>, Long> kTable = kStream.flatMapValues(value -> Arrays.asList(value.split(" ")))

                    //根据value的值进行分组
                    .groupBy((k, v) -> v)

                    /*设置每个窗口与每个窗口之间的时间间隔,同时设置推进的时间间隔
                    例：0-10s
                      5-15s
                      10-20s
                    */
                    .windowedBy(SessionWindows.with(Duration.ofSeconds(10)))

                    .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("AA").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));

            kStream.flatMapValues(value -> Arrays.asList(value.split(" ")))

                    //根据value的值进行分组
                    .groupBy((k, v) -> v)


                    //设置每个窗口与每个窗口之间的时间间隔
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))

                    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("AA").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));


            kTable.toStream().foreach((k,v)->{

                //获取window对象
                Window window = k.window();

                //获取时间间隔的开始时间
                long start = window.start();

                //获取时间间隔的结束时间
                long end = window.end();

                //时间格式化的处理
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String startTime = dateFormat.format(new Date(start));
                String endTime = dateFormat.format(new Date(end));

                System.out.println(startTime+"\t"+endTime+"\t"+k.key()+"\t"+v);


            });


            //获取topology任务对象
            Topology topology = streamsBuilder.build();

            //创建kafkaStream对象启动任务
            KafkaStreams kafkaStreams = new KafkaStreams(topology,properties);

            //开启任务
            kafkaStreams.start();


        }





}
